package com.shujia.iteminfo

import com.shujia.utils.{Constants, SparkMain, SparkTool}
import org.apache.spark.sql.{DataFrame, Row, SparkSession}
import redis.clients.jedis.Jedis

object ItemInfoLabelJoin extends SparkMain {
  val spark: SparkSession = SparkTool.getSparkSession(this.getClass.getName.replace("$", ""))

  import spark.implicits._
  import org.apache.spark.sql.functions._

  override def run(day: String): Unit = {

    // 加载媒资库数据
    val mediaDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "#")
      .schema("id STRING,name STRING,category STRING,tags STRING,actors STRING,directors STRING,hot STRING,score STRING,releasedate STRING,similar STRING,area STRING,language STRING,isnew STRING,mapid STRING,duration STRING,hottags STRING,coldtags STRING")
      .load(Constants.MEDIA_SOURCE_PATH)

    val labelList: List[String] = List("actors", "area", "category", "directors", "language", "tags")


    for (labelName <- labelList) {
      getLabelJoin(mediaDF, labelName)
    }


  }

  def getLabelJoin(mediaDF: DataFrame, labelName: String): Unit = {
    // 加载Label中的数据
    val labelDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", "#")
      .schema("labelId STRING,labelName STRING,code STRING,date STRING")
      .load(s"${Constants.ITEM_LABEL_BASE_PATH}/$labelName")

    mediaDF
      .withColumn("labelId", explode(split($"$labelName", ",")) as "labelName")
      .join(labelDF, "labelId")
      .groupBy("id", "name")
      .agg(collect_set("labelName") as "labelNames")
      .select($"id", $"name", concat_ws(",", $"labelNames") as s"$labelName")
      .foreachPartition(iter => {
        val jedis: Jedis = new Jedis(Constants.REDIS_HOST, Constants.REDIS_PORT)
        iter.foreach {
          case Row(id: String, name: String, actors: String) =>
            // 以影片的id+影片的name作为key ”label“作为field actors作为value
            jedis.hset(s"item_info:$id-$name", s"$labelName", actors)

            // 设置过期时间 自动清理数据
            jedis.expire(s"item_info:$id-$name", 24 * 60 * 60)

        }
      })

  }


}
